#pragma once
#include <asm-generic/socket.h>
#include <cerrno>
#include <condition_variable>
#include <csignal>
#include <cstdint>
#include <cstdlib>
#include <ctime>
#include <iostream>
#include <functional>
#include <sys/time.h>
#include <unordered_map>
#include <string>
#include <thread>
#include <mutex>
#include <vector>
#include <memory>
#include <sys/types.h>
#include <assert.h>
#include <string.h>
#include <sys/types.h>        
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <signal.h>
#include "detail.hpp"

#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{
public:
    Buffer() : _read_idx(0), _write_idx(0), _buffer(BUFFER_DEFAULT_SIZE)
    {}
    char* Begin() {return &*_buffer.begin();}
    //获取当前写入起始位置
    char* WritePosition() {return Begin() + _write_idx;} 
    //获取当前读取起始位置
    char* ReadPosition() {return Begin() + _read_idx;} 
    //获取缓冲区末尾空闲空间大小--写偏移之后的空闲时间
    uint64_t TailIdleSize() {return _buffer.size() - _write_idx;}
    //获取缓冲区起始空闲空间大小--读偏移之前的空闲空间
    uint64_t HeadIdleSize() {return _read_idx;}
    //获取可读数据大小
    const uint64_t ReadAbleSize() const {return _write_idx - _read_idx;} 
    //将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        if(len == 0)    return;
        assert(len <= ReadAbleSize());
        _read_idx += len;
    }
    //将写偏移向后移动
    void MoveWriteOffset(uint64_t len)
    {
        //移动的大小，必须小于当前后面的空间大小
        assert(len <= TailIdleSize());
        _write_idx += len;
    }
    //确保可写空间足够 (整体空闲空间足够就移动数据，否则就扩容)
    void EnsureWriteSpace(uint64_t len)
    {
        //末尾空间足够，直接返回
        if(len <= TailIdleSize())
            return;
        //末尾空间不够，考虑加上起始位置大小够不够，够则移动到前面来
        if(len <= TailIdleSize() + HeadIdleSize())
        {
            //将数据移动到起始位置
            uint64_t rsz = ReadAbleSize();
            std::copy(ReadPosition(), ReadPosition()+ rsz, Begin());
            _read_idx = 0;
            _write_idx = rsz;
        }
        else 
        {
            //总体空间不够，直接扩容足够空间即可
            _buffer.resize(_write_idx+len);
        }
    }
    //写入数据
    void Write(const void *data, uint64_t len)
    {
        EnsureWriteSpace(len);
        const char *d = (const char*)data;
        std::copy(d, d+len, WritePosition());
    }
    void WriteAndPush(const void *data, uint64_t len)
    {
        Write(data, len);
        return MoveWriteOffset(len);
    }
    void WriteString(const std::string &data)
    {
        return Write(data.c_str(), data.size());
    }
    void WriteStringAndPush(const std::string &data)
    {
        WriteString(data);
        return MoveWriteOffset(data.size());
    }
    void WriteBuffer(Buffer &data) 
    {
        return Write(data.ReadPosition(), data.ReadAbleSize());
    }
    void WriteBufferAndPush(Buffer &data) 
    {
        WriteBuffer(data);
        return MoveWriteOffset(data.ReadAbleSize());
    }
    //读取数据
    void Read(void *buf, uint64_t len)
    {
        //要求获取的数据必须小于可读数据大小
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(), ReadPosition()+len, (char*)buf);
    }
    void ReadAndPop(void *buf, uint64_t len)
    {
        Read(buf, len);
        return MoveReadOffset(len);
    }
    std::string ReadAsString(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    std::string ReadAsStringAndPop(uint64_t len)
    {
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }
    //找到换行字符
    const void* findCRLF() 
    {
        const void *res =  memchr(ReadPosition(), '\n', ReadAbleSize());
        return res;
    }
    std::string GetLine()
    {
        const char *pos = (const char*)findCRLF();
        if(pos == NULL)
            return "";
        //+1为了把\n也取出来
        return ReadAsString(pos-ReadPosition()+1);
    }
    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }
    //清空缓冲区
    void Clear()
    {
        _read_idx = 0;
        _write_idx = 0;
    }
private:
    std::vector<char> _buffer;
    uint64_t _read_idx;     //读偏移
    uint64_t _write_idx;    //写偏移
};

const static int max_listen = 1024;
class Socket
{
public:
    Socket():_sockfd(-1)
    {}
    Socket(int fd):_sockfd(fd)
    {}
    ~Socket() { Close();}
    int Fd() {return _sockfd;}
    //创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if(_sockfd < 0)
        {
            ERR_LOG("socket create failed!");
            return false;
        }
        return true;
    }
    //绑定地址信息
    bool Bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = bind(_sockfd, (struct sockaddr*)&addr, len);
        if(ret < 0)
        {
            ERR_LOG("socket bind address failed!");
            return false;
        }
        return true;
    }
    //开始监听
    bool Listen(int backlog = max_listen)
    {
        int ret = listen(_sockfd, backlog);
        if(ret < 0)
        {
            ERR_LOG("socket listen failed!");
            return false;
        }
        return true;
    }
    //向服务器发起连接
    bool Connect(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = connect(_sockfd, (struct sockaddr*)&addr, len);
        if(ret < 0)
        {
            ERR_LOG("socket connect failed!");
            return false;
        }
        return true;
    }
    //获取新连接
    int Accept()
    {
        int newfd = accept(_sockfd, nullptr, nullptr);
        if (newfd < 0)
        {
            ERR_LOG("socket accept failed!");
            return -1;
        }
        return newfd;
    }
    //接收数据
    ssize_t Recv(void *buf, size_t len, int flag = 0)
    {
        ssize_t ret = recv(_sockfd, buf, len, flag);
        if(ret <= 0)
        {
            //EAGAIN : 当前socket接收缓冲区没有数据，非阻塞情况下会有这个错误
            //EAINTR ：当前socket阻塞等待，被信号打断了
            if(errno == EAGAIN || errno == EINTR)
                return 0;   //表示没有接收到数据
            ERR_LOG("socket recv failed!");
            return -1;
        }
        return ret; // 实际接受到的数据长度
    }
    ssize_t NonBlockRecv(void *buf, size_t len, int flag = 0)
    {
        //非阻塞版本
        return Recv(buf, len, MSG_DONTWAIT);
    }
    //发送数据
    ssize_t Send(const void *buf, size_t len, int flag = 0)
    {
        ssize_t ret = send(_sockfd, buf, len, flag);
        if(ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR) 
            {
                return 0;
            }
            ERR_LOG("socket send failed!");
            return -1;
        }
        return ret; //实际发送的数据长度
    }
    ssize_t NonBlockSend(const void *buf, size_t len)
    {
        if(len == 0) return 0;
        //非阻塞版本
        return Send(buf, len, MSG_DONTWAIT);
    }
    //关闭套接字
    void Close() 
    { 
        if(_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;  
        }
    }
    //创建一个服务器连接
    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false)
    {
        //1创建套接字， 2.绑定地址， 3.开始监听, 4.设置非阻塞， 5.启动地址复用
        if(!Create()) return false;
        if(block_flag) NonBlock();
        if(!Bind(ip, port)) return false;
        if(!Listen()) return false;
        ReuseAddress();
        return true;
    }
    //创建一个客户端连接
    bool CreateClient(uint16_t port, const std::string &ip)
    {
        //创建套接字， 连接指定服务器
        if(!Create()) return false;
        if(!Connect(ip, port)) return false;
        return true;
    }
    //设置套接字选项--开启地址&&端口重用
    void ReuseAddress()
    {
        int val = 1;
        //setsockopt(int fd, int level, int optname, const void *optval, socklen_t optlen)
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
    }
    //设置套接字阻塞属性--非阻塞
    void NonBlock()
    {
        //获取当前属性
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
private:
    int _sockfd;
};

class Poller;
class EventLoop;
class Channel
{
    using EventCallBack = std::function<void()>;    
public:
    Channel(EventLoop *loop, int fd):_fd(fd), _events(0), _revents(0), _loop(loop)
    {}
    int Fd() {return _fd;}
    uint32_t Events() {return _events;} //获取想要监控的事件
    void SetREvents(uint32_t events) {_revents = events;} //设置实际监控的事件
    void SetReadCallBack(const EventCallBack &cb) {_read_callback = cb;}
    void SetWriteCallBack(const EventCallBack &cb) {_write_callback = cb;}
    void SetErrorCallBack(const EventCallBack &cb) {_error_callback = cb;}
    void SetCloseCallBack(const EventCallBack &cb) {_close_callback = cb;}
    void SetEventCallBack(const EventCallBack &cb) {_event_callback = cb;}
    //当前是否监控了可读
    bool ReadAble() {return (_events & EPOLLIN);}   
    //当前是否监控了可写
    bool WriteAble() {return (_events & EPOLLOUT);}
    //启动读事件监控
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }
    //启动写事件监控
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Update();
    }
    //关闭读事件监控
    void DisableRead()
    {
        _events &= (~EPOLLIN);
        Update();
    }
    //关闭写事件监控
    void DisableWrite()
    {
        _events &= (~EPOLLOUT);
        Update();
    }
    //关闭所有事件监控
    void DisableAll()
    {
        _events = 0;
        Update();
    }
    //移除监控
    void Remove();
    void Update();

    //事件处理，一旦连接出发了事件，就调用这个函数
    void HandleEvent()
    {
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            //不管任何事件，都调用的回调函数
            if(_event_callback) _event_callback();
            if(_read_callback) _read_callback();
        }
        //有可能会释放连接的操作，一次只处理一个
        if(_revents & EPOLLOUT)
        {
            //不管任何事件，都调用的回调函数
            if(_event_callback) _event_callback();
            if(_write_callback) _write_callback();
        }
        else if(_revents & EPOLLERR)
        {
            if(_event_callback) _event_callback();
            if(_error_callback) _error_callback(); //一旦出错，就会释放连接，就没必要调用任意回调了，因此放到前面回调
        }
        else if(_revents & EPOLLHUP)
        {
            if(_event_callback) _event_callback();
            if(_close_callback) _close_callback();
        }

    }
private:
    int _fd;
    EventLoop *_loop;
    uint32_t _events;   //当前需要监控的事件
    uint32_t _revents;  //当前连接的事件
    EventCallBack _read_callback;     //可读事件被触发的回调
    EventCallBack _write_callback;    //可写事件被触发的回调
    EventCallBack _error_callback;    //错误事件被触发的回调
    EventCallBack _close_callback;    //任意断开事件被触发的回调
    EventCallBack _event_callback;    //任意时间事件被触发的回调
};

#define MAX_EPOLLEVENTS 1024
class Poller
{
public:
    Poller()
    {
        _epfd = epoll_create(MAX_EPOLLEVENTS);
        if (_epfd < 0)
        {
            ERR_LOG("EPOLL CREATE FAILED!");
            abort();
        }
    }
    void UpdateEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if(ret == false)
        {
            //不存在则添加
            _channels.insert({channel->Fd(), channel});
            return Update(channel, EPOLL_CTL_ADD);
        }
        return Update(channel, EPOLL_CTL_MOD);
    }
    void RemoveEvent(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if(it != _channels.end())
        {
            _channels.erase(it);
        }
        Update(channel, EPOLL_CTL_DEL);
    }
    //开始监控，返回活跃连接
    void Poll(std::vector<Channel*> *active)
    {
        //epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)
        //timeout = -1表示阻塞监控
        //返回就绪事件个数
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
        if(nfds < 0)
        {
            if(errno == EINTR)
                return;
            ERR_LOG("EPOLL WAIT ERROR : %s", strerror(errno));
            abort();
        }
        for(int i = 0; i < nfds; i++)
        {
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            it->second->SetREvents(_evs[i].events);
            active->push_back(it->second);
        }
        return;
    }
private:
    //对epoll的直接操作
    void Update(Channel *channel, int op)
    {
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();
        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if(ret < 0)
        {
            ERR_LOG("EPOLLCTL FAILED!");
        }
        return;
    }
    //判断一个Channel是否已经添加了事件监控
    bool HasChannel(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if(it == _channels.end())
        {
            return false;
        }
        return true;
    }
private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    std::unordered_map<int, Channel*> _channels;
};

class TimerTask
{
public:
    using TaskFunc = std::function<void()>; 
    using ReleaseFunc = std::function<void()>;
    TimerTask(uint64_t id, uint32_t delay, const TaskFunc &task):_id(id), _timeout(delay), _task(task), _canceled(false)
    {}
    ~TimerTask() 
    {
        if (!_canceled) _task();
        _release();
    }
    void setRealease(const ReleaseFunc &release) {_release = release;}
    uint32_t DelayTime() {return _timeout;}
    void Cancel() {_canceled = true;}
private:
    uint64_t _id; //定时任务对象id
    uint32_t _timeout; //定时任务的超时时间
    TaskFunc _task; //定时器对象执行的定时任务
    ReleaseFunc _release; //删除TimerWheel保存的定时器对象信息
    bool _canceled;
};

class TimerWheel
{
    using WeakTask = std::weak_ptr<TimerTask>;
    using PtrTask = std::shared_ptr<TimerTask>;
public:
    TimerWheel(EventLoop* loop):_ticket(0), _capacity(60), _wheel(_capacity)
    , _loop(loop), _timerfd(CreateTimerfd()), _timer_channel(std::make_unique<Channel>(_loop, _timerfd))
    {
        _timer_channel->SetReadCallBack(std::bind(&TimerWheel::OnTime, this));
        _timer_channel->EnableRead();
    }
    //把该函数放置EventLoop模块中执行，以保证线程安全
    void TimerAdd(const uint64_t id, const uint32_t delay, const TimerTask::TaskFunc &cb);
    void TimerReflash(const uint64_t id);
    void TimerCancel(const uint64_t id);

    //该接口存在线程安全问题，只能在模块内，对应的EventLoop中执行
    bool HasTimer(const uint64_t id)
    {
        auto it = _times.find(id);
        if(it == _times.end())
        {
            //没找到对应任务，直接返回
            return false;
        }
        return true;
    }
private:
    void RemoveTimer(uint64_t id)
    {
        auto it = _times.find(id);
        if(it != _times.end())
            _times.erase(it);
    }
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if(timerfd < 0)
        {
            ERR_LOG("timerfd create failed!");
            abort();
        }
        static itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0; //第一次超时时间为1s后
        itime .it_interval.tv_sec = 1;
        itime .it_interval.tv_nsec = 1; //第一次超时后 ，每次超时的间隔时间
        timerfd_settime(timerfd, 0, &itime, nullptr);
        return timerfd;
    }
    void ReadTimerfd()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, 8);
        if(ret < 0)
        {
            ERR_LOG("read timerfd failed!");
            abort();
        }
        return;
    }
    //应每秒被执行一次
    void RunTimerTask()
    {
        _ticket = (_ticket + 1) % _capacity;
        _wheel[_ticket].clear();   //执行析构
    }
    void OnTime()
    {
        ReadTimerfd();
        RunTimerTask();
    }
    void TimerAddInLoop(const uint64_t id, const uint32_t delay, const TimerTask::TaskFunc &cb)   //添加定时任务
    {
        PtrTask pt(new TimerTask(id, delay, cb));
        pt->setRealease(std::bind(&TimerWheel::RemoveTimer, this, id));
        int pos = (delay + _ticket) % _capacity;
        _wheel[pos].emplace_back(pt);
        _times.insert({id, WeakTask(pt)});
    }
    void TimerReflashInLoop(uint64_t id)  //刷新定时任务
    {
        auto it = _times.find(id);
        if(it == _times.end())
        {
            //没找到对应任务，直接返回
            return;
        }
        PtrTask pt = it->second.lock(); //获取weak_ptr管理的share_ptr
        int delay = pt->DelayTime();
        int pos = (delay + _ticket) % _capacity;
        _wheel[pos].emplace_back(pt);
    }
    void TimerCancelInLoop(const uint64_t &id)
    {
        auto it = _times.find(id);
        if(it == _times.end())
        {
            //没找到对应任务，直接返回
            return;
        }
        PtrTask pt = it->second.lock(); 
        if(pt)  pt->Cancel();
    }
private:
    int _ticket;    //秒表，走到哪释放哪
    int _capacity;  //最大延迟时间
    std::vector<std::vector<PtrTask>> _wheel;
    std::unordered_map<uint64_t, WeakTask> _times;
    EventLoop *_loop;
    int _timerfd; //定时器描述符--可读事件回调是读取计数器，执行定时任务
    std::unique_ptr<Channel> _timer_channel;
};


class EventLoop
{
public:
    using Functor = std::function<void()>;
public:

    EventLoop():_event_fd(CreateEventFd()), 
        _thread_id(std::this_thread::get_id()), 
        _event_channel(std::make_unique<Channel>(this, _event_fd)),
        _timer_wheel(this)
    {
        //给EventFd添加可读事件回调，读取eventfd事件通知次数
        _event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEventFd, this));
        _event_channel->EnableRead();
    }
    //判断将要执行的任务是否处于当前线程中，是则执行，否则压入任务池
    void RunInLoop(const Functor &cb)
    {
        if(IsInLoop())
            return cb();
        else
            return QueueInLoop(cb);
    }
    //压入任务池的操作
    void QueueInLoop(const Functor &cb)
    {
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _tasks.push_back(cb);
        }
        //唤醒有可能因为没有事件就绪而导致的epoll阻塞
        WakeUpEventFd();
    }
    //判断当前线程是否是EventLoop对应的线程
    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }
    void AssertInLoop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }
    //添加/修改描述符的事件监控
    void UpdateEvent(Channel *channel)
    {
        return _poller.UpdateEvent(channel);
    }
    //移除描述符监控
    void RemoveEvent(Channel *channel)
    {
        return _poller.RemoveEvent(channel);
    }
    //事件监控->就绪事件处理->执行任务
    void Start()
    {
        while(1)
        {
            //1
            std::vector<Channel*> actives;
            _poller.Poll(&actives);
            //2
            for(auto &channel : actives)
            {
                channel->HandleEvent();
            }
            //3
            RunAllTask();
        }
    }
    void TimerAdd(uint64_t id, uint32_t delay, const TimerTask::TaskFunc &cb)
    {
        _timer_wheel.TimerAdd(id, delay, cb);
    }
    void TimerReflash(const uint64_t id)
    {
        return _timer_wheel.TimerReflash(id);
    }
    void TimerCancel(const uint64_t id)
    {
        return _timer_wheel.TimerCancel(id);
    }
    bool HasTimer(const uint64_t id)
    {
        return _timer_wheel.HasTimer(id);
    }
private:
    void RunAllTask()
    {
        std::vector<Functor> functor;

        {
            std::unique_lock<std::mutex> lock(_mtx);
            _tasks.swap(functor);
        }
        for(auto &func : functor)
        {
            func();
        }
        return;
    }
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if(efd < 0)
        {
            ERR_LOG("eventfd create failed!");
            abort();
        }
        return efd;
    }
    void ReadEventFd()
    {
        uint64_t val = 1;
        int ret = read(_event_fd, &val, sizeof(val));
        if(ret < 0)
        {
            //EINTR: 被信号打断； EAGAIN: 无数据可读
            if(errno == EINTR || errno == EAGAIN)
                return;
            ERR_LOG("read eventfd failed!");
            abort();
        } 
        return;
    }      
    void WakeUpEventFd()
    {
        //往eventfd写入数据，就触发eventfd写事件了，成功唤醒
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(val));
        if(ret < 0)
        {
            if(errno == EINTR)
                return;
            ERR_LOG("wake eventfd up failed!");
            abort();
        } 
        return;
    }
private:
    std::thread::id _thread_id; //线程ID
    int _event_fd; //eventfd唤醒IO事件有可能导致的阻塞
    std::unique_ptr<Channel> _event_channel;
    Poller _poller; //进行所有描述符的事件监控
    std::vector<Functor> _tasks; //任务池
    std::mutex _mtx;
    TimerWheel _timer_wheel;
};

class LoopThread
{
public:
    //创建线程，设定线程入口函数
    LoopThread(): _loop(nullptr), _thread(&LoopThread::ThreadEntry, this)
    {}
    //返回当前线程关联的EventLoop对象指针
    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;

        {
            //表示必须_loop不等于空才会被唤醒，否则该任务一直被阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]() {return _loop != nullptr;});
            //此时_loop已经实例化完成
            loop = _loop;
        }

        return loop;
    }
private:
    //实例化EventLoop对象，并且开始运行EventLoop模块的功能
    void ThreadEntry()
    {
        //不给_loop在堆上是希望loop的生命周期在该线程内部，随着线程销毁而销毁
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        _loop->Start();
    }
private:
    //用于实现——loop获取的同步关系,避免线程创建了，在_loop创建之前去获取_loop
    std::mutex _mutex;
    std::condition_variable _cond;
    EventLoop *_loop;       //EventLoop指针变量，需要在线程内实例化
    std::thread _thread;    //EventLoop对应的线程
};

class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop *baseloop): _thread_count(0), _next_idx(0), _baseloop(baseloop)
    {}
    //设置线程数量
    void SetThreadCount(int count) { _thread_count = count;}
    //设置所有从属线程
    void Create()
    {
        if (_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for(int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                //如果当前EventLoop还没有实例化完成，则会阻塞
                _loops[i] = _threads[i]->GetLoop();
            }
        }
        return;
    }
    EventLoop *NextLoop()
    {
        if(_thread_count == 0)
            return _baseloop;
        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }
private:
    int _thread_count;      //从属线程数量
    int _next_idx;     
    EventLoop *_baseloop;   //主EventPoll,运行在主线程，从属线程为0，则操作都在baseloop中
    std::vector<LoopThread*> _threads;  //保存所有LoopThread对象
    std::vector<EventLoop*> _loops;     //从属线程大于0则从当前容器中进行线程eventloop分配
};

class Acceptor
{
    using AcceptCallback = std::function<void(int)>;
public:
    //设置读事件回调，必须放在启动读事件监听后，否则可能启动监控后，立即有事件，此时回调还未设置，连接得不到处理，资源泄漏
    Acceptor(EventLoop* loop, int port):_loop(loop), _channel(loop, _socket.Fd()), _socket(CreateServer(port))
    {
        _channel.SetReadCallBack(std::bind(&Acceptor::HandleRead, this));
    }
    
    void SetAcceptCallback(const AcceptCallback &cb) { _accept_cb = cb;}
    void Listen() {_channel.EnableRead();}
private:
    //监听套接字的回调函数处理函数--获取新连接，调用accept_cb函数进行新连接处理
    void HandleRead()
    {
        int newfd = _socket.Accept();
        if(newfd < 0)
        {
            ERR_LOG("Acceptor: create Accept failed!");
            return;
        }
        if (_accept_cb) _accept_cb(newfd);
    }
    int CreateServer(int port)
    {
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }
private:
    Socket _socket; //创建监听套接字
    EventLoop *_loop; //用于对监听套接字进行事件监控
    Channel _channel; //用于对监听套接字进行时间管理

    AcceptCallback _accept_cb;
};



typedef enum 
{
    DISCONNECTED,   //连接关闭状态
    CONNECTING,     //连接成功，待处理状态
    CONNECTED,      //连接成功，设置完成，可以通信状态
    DISCONNECTING   //待关闭状态
}ConnStatu;

class Connection : public std::enable_shared_from_this<Connection>
{
public:
    using Ptr = std::shared_ptr<Connection>;
private:
    //让服务器模块设置的
    using ConnectedCallback = std::function<void(const Connection::Ptr&)>;
    using MessageCallback = std::function<void(const Connection::Ptr&, Buffer*)>;
    using ClosedCallback = std::function<void(const Connection::Ptr&)>;
    using AnyEventCallback = std::function<void(const Connection::Ptr&)>;
public:
    Connection(EventLoop* loop, uint64_t conn_id, int sockfd):_conn_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false), 
        _loop(loop), _socket(sockfd),_statu(CONNECTING), _channel(_loop, sockfd)
    {
        _channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));
        _channel.SetEventCallBack(std::bind(&Connection::HandleEvent, this));
        _channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));
        _channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));
    }
    ~Connection()
    {   
        DBG_LOG("release connection: %p", this);
    }
    int Fd() {return _sockfd;}
    int Id() {return _conn_id;}
    bool Connected() {return _statu == CONNECTED;}
    void SetContext(const Any &context) {_context = context;}
    Any* GetContext() {return &_context;}
    void SetConnectedCallBack(const ConnectedCallback &cb) {_connected_callback = cb;}
    void SetMessageCallBack(const MessageCallback &cb) {_message_callback = cb;}
    void SetClosedCallBack(const ClosedCallback &cb) {_closed_callback = cb;}
    void SetAnyEventCallBack(const AnyEventCallback &cb) {_event_callback = cb;}
    void SetSrvClosedCallback(const ClosedCallback &cb) {_server_close_callback = cb;}
    //连接建立就绪后，进行channel回调设置，启动读监控，调用connected_cb
    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    //发送数据，将数据放回到发送缓冲区，启动写事件监控
    void Send(const char *data, size_t len)
    {
        Buffer buff;
        buff.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, buff));

    }
    //提供给组件使用者的关闭接口，并不实际关闭，需要判断有无数据待处理
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }
    // void Release()
    // {
    //     _loop->QueueInLoop(std::bind(&Connection::Release, this));
    // }
    //启动非活跃销毁，定义多长时间无通信为非活跃，添加定时任务
    void EnableInactiveRelease(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    //取消非活跃销毁
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    //切换协议--重置上下文以及阶段性处理函数--必须在EventLoop线程中立即执行
    void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg,
    const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context,conn, msg, closed, event));
    }
private:
    //五个channel事件回调函数
    //接受socket数据放到接受缓冲区，调用message_cb
    void HandleRead()
    {
        //接收socket数据，放入缓冲区
        char buff[65536];
        ssize_t  ret = _socket.NonBlockRecv(buff, 65535);
        if (ret < 0)
        {
            //出错，不能直接关闭连接，修改状态为待关闭
            return ShutdownInLoop();
        }
        //将数据放入输入缓冲区
        _in_buffer.WriteAndPush(buff, ret);
        //调用message_cb
        if(_in_buffer.ReadAbleSize() > 0)
        {
            //shared_from_this:从当前对象自身获取自身的shared_ptr管理对象
            return _message_callback(shared_from_this(), &_in_buffer);
        }
    }
    //将发送缓冲区中的数据发送
    void HandleWrite()
    {
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        if(ret < 0)
        {
            //关闭连接
            if(_in_buffer.ReadAbleSize() > 0)
            {
                _message_callback(shared_from_this(), &_in_buffer);
            }
            return ReleaseInLoop(); //实际关闭连接
        }
        //将读数据向后偏移
        _out_buffer.MoveReadOffset(ret);
        if(_out_buffer.ReadAbleSize() == 0)
        {
            _channel.DisableWrite();
            //如果当前是连接带关闭状态，有数据则发完数据释放，无则直接释放
            if(_statu == DISCONNECTING)
            {
                return ReleaseInLoop();
            }
        }
        return;
    }
    void HandleClose()
    {
        //一旦连接挂断，套接字就干不了了，因此有数据待处理一下就处理，完毕关闭连接
        if(_in_buffer.ReadAbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        return ReleaseInLoop(); //实际关闭连接
    }
    void HandleError()
    {
        return HandleClose();
    }
    //刷新连接活跃度，调用组件使用者的任意时间
    void HandleEvent()
    {
        if(_enable_inactive_release == true)
        {
            _loop->TimerReflash(_conn_id);
        }
        if(_event_callback)
        {
            _event_callback(shared_from_this());
        }
    }
    //连接获取之后，所处的状态下进行各种设置
    void EstablishedInLoop()
    {
        //1.修改连接状态
        assert(_statu == CONNECTING); //当前状态一定是上层半连接状态
        _statu = CONNECTED; //进入已完成连接状态
        //一旦启动读时间监控可能会立即出发读事件
        _channel.EnableRead();
        if(_connected_callback)
            _connected_callback(shared_from_this());
    }
    //实际的释放接口
    void ReleaseInLoop()
    {
        //修改连接状态
        _statu = DISCONNECTED;
        //移除连接的事件监控
        _channel.Remove();
        //关闭描述符
        _socket.Close();
        //如果当时定时器队列中还有定时销毁任务，则取消
        if(_loop->HasTimer(_conn_id))
            CancelInactiveReleaseInLoop();
        //调用关闭回调函数
        if(_closed_callback)
            _closed_callback(shared_from_this());
        //移除服务器内部管理信息
        if(_server_close_callback)
            _server_close_callback(shared_from_this());
    }
    //把数据放入发送缓冲区，启动可写事件监控
    void SendInLoop(Buffer buff)
    {
        if(_statu == DISCONNECTED)  return;
        _out_buffer.WriteBufferAndPush(buff);
        if(_channel.WriteAble() == false)
            _channel.EnableWrite();
    }
    //需要判断是否还有数据待处理，待发送
    void ShutdownInLoop()
    {
        _statu = DISCONNECTING; //半关闭状态
        if(_in_buffer.ReadAbleSize() > 0)
        {
            if(_message_callback)   _message_callback(shared_from_this(), &_in_buffer);
        }
        if(_out_buffer.ReadAbleSize() > 0)
        {
            if(_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }
        if(_out_buffer.ReadAbleSize() == 0)
        {
            ReleaseInLoop();
        }
    }
    //启动非活跃连接超时释放规则
    void EnableInactiveReleaseInLoop(int sec)
    {
        //将判断标志 _enable_inactive_release 置为true
        _enable_inactive_release = true;
        //如果当时定时销毁任务已经存在，则刷新延迟
        if(_loop->HasTimer(_conn_id))
            return _loop->TimerReflash(_conn_id);
        //反则新增
        _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::ReleaseInLoop, this));
    }
    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if(_loop->HasTimer(_conn_id))
        {
            _loop->TimerCancel(_conn_id);
        }
    }
    //切换协议--重置上下文以及阶段性回调处理函数 -- 非线程安全
    void UpgradeInLoop(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg,
    const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }
private:
    uint64_t _conn_id;      //连接的唯一ID，便于连接的管理和查找
    //uint64_t _timer_id    //定时器ID，具有唯一性，为简化操作使用_conn_id代替
    int _sockfd;            //连接关联的文件描述符
    bool _enable_inactive_release; //连接是否启动非活跃销毁的判断标志，default = false
    EventLoop *_loop;
    ConnStatu _statu;
    Socket _socket;         //套接字操作管理
    Channel _channel;       //连接的事件管理
    Buffer _in_buffer;      //输入缓冲区--存放从socket中读到的数据
    Buffer _out_buffer;     //输出缓冲区--存放要发送给对端的数据
    Any _context;

    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
    //组件内的连接关闭回调，因为服务器组件内会把所有连接管理起来，一旦某个连接关闭，
    //就应该从管理的地方移除自己的信息
    ClosedCallback _server_close_callback;
};

class TcpServer
{
    using ConnectedCallback = std::function<void(const Connection::Ptr&)>;
    using MessageCallback = std::function<void(const Connection::Ptr&, Buffer*)>;
    using ClosedCallback = std::function<void(const Connection::Ptr&)>;
    using AnyEventCallback = std::function<void(const Connection::Ptr&)>;
public:
    TcpServer(int port):_port(port), _next_id(0), _enable_inactive_release(false),
        _acceptor(&_baseloop, _port), _pool(&_baseloop)
    {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen(); //将监听套接字挂到baseloop上开始监控事件
    }
    void SetThreadCount(int count)  { return _pool.SetThreadCount(count);}
    void SetConnectedCallBack(const ConnectedCallback &cb) {_connected_callback = cb;}
    void SetMessageCallBack(const MessageCallback &cb) {_message_callback = cb;}
    void SetClosedCallBack(const ClosedCallback &cb) {_closed_callback = cb;}
    void SetAnyEventCallBack(const AnyEventCallback &cb) {_event_callback = cb;}
    void EnableInactiveRelease(int timeout) 
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    void RunAfter(const EventLoop::Functor &task, int delay)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));

    }
    void Start()
    {
        _pool.Create(); //创建从属线程
        return _baseloop.Start();
    }
private:
    void RunAfterInLoop(const EventLoop::Functor &task, int delay)
    {
        _next_id++;
        _baseloop.TimerAdd(_next_id,  delay, task);
    }
    //为新连接构建Connection进行管理
    void NewConnection(int fd)
    {
        _next_id++;
        Connection::Ptr conn(new Connection(_pool.NextLoop(), _next_id, fd));
        conn->SetMessageCallBack(_message_callback);
        conn->SetConnectedCallBack(_connected_callback);
        conn->SetAnyEventCallBack(_event_callback);
        conn->SetClosedCallBack(_closed_callback);
        conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if(_enable_inactive_release)
            conn->EnableInactiveRelease(_timeout);
        conn->Established(); //就绪初始化
        _conns.insert({_next_id, conn});
    }
    void RemoveInLoop(const Connection::Ptr &conn)
    {
        int id = conn->Id();
        auto it = _conns.find(id);
        if(it == _conns.end())
        {
            ERR_LOG("conn_id not found!");
            return;
        }
        _conns.erase(it);
    }
    
    //从管理Connection的conns中移除连接信息
    void RemoveConnection(const Connection::Ptr &conn)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveInLoop, this, conn));
    }
private:
    int _port;
    uint64_t _next_id;      //自动增长的连接id
    int _timeout;           //非活跃连接的统计时间--多长事件无通信就是非活跃连接
    bool _enable_inactive_release;  //是否启动了非活跃连接超市销毁的判断标志
    EventLoop _baseloop;    //主线程的EventLoop对象，负责监听事件的处理
    Acceptor _acceptor;     //这是监听套接字的管理对象
    LoopThreadPool _pool;   //从属EventPool的线程池
    std::unordered_map<uint64_t, Connection::Ptr> _conns;   //保存管理所有连接对应的shared_ptr对象
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
};


inline void Channel::Remove()
{
    return _loop->RemoveEvent(this);
}
inline void Channel::Update()
{
    return _loop->UpdateEvent(this);
}

inline void TimerWheel::TimerAdd(const uint64_t id, const uint32_t delay, const TimerTask::TaskFunc &cb)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, delay, cb));
}
inline void TimerWheel::TimerReflash(const uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerReflashInLoop, this, id));
}
inline void TimerWheel::TimerCancel(const uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
}

struct NetWork
{
    NetWork()
    {
        signal(SIGPIPE, SIG_IGN);
    }
};

static struct NetWork nw;